package com.project.pulsar.messageModel;

import com.project.pulsar.conf.PulsarConf;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * 异步、同步消息接收与发送
 * 消息批量消费
 */
@Slf4j
@RestController
public class MessageModelController {
    @Autowired
    PulsarConf pulsarConf;


    /**
     * 同步-生产消息
     * @param msg
     * @throws PulsarClientException
     */
    @GetMapping("/model/sendMsgBySync")
    public MessageId sendMsgBySync(String msg) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();

        Producer<byte[]> producer = pulsarFactory.newProducer()
                .topic("my-topic")
                .create();
        // 然后你就可以发送消息到指定的broker 和topic上：
        return producer.send(msg.getBytes());
    }
    /**
     * 异步-生产消息
     * @param msg
     * @throws PulsarClientException
     */
    @GetMapping("/model/sendMsgAsync")
    public void sendMsgAsync(String msg) throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();

        Producer<byte[]> producer = pulsarFactory.newProducer()
                .topic("my-topic")
                .create();
        // 然后你就可以发送消息到指定的broker 和topic上：

        producer.sendAsync(msg.getBytes()).thenAccept(msgId -> {
            System.out.println("Message with ID " + msgId + " successfully sent");
        });
    }

    /**
     * 手动执行同步获取消息
     * @throws PulsarClientException
     */
    @GetMapping("/model/comsumerBySync")
    public void comsumerByArtificialBySync() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        System.out.println(new String(receive.getData()));
        consumer.acknowledge(receive);//确认消息被消费
        consumer.close();
    }

    /**
     * 手动执行异步获取消息
     * @throws PulsarClientException
     */
    @GetMapping("/model/comsumerByASync")
    public void comsumerByArtificialByASync() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .subscribe();
        CompletableFuture<Message<byte[]>> asyncMessage = consumer.receiveAsync();
        asyncMessage.thenAccept(message -> {
            System.out.println(new String(message.getData()));
            try {
                MessageId messageId = message.getMessageId();
                consumer.acknowledge(messageId);
                consumer.close();
            } catch (PulsarClientException e) {
                e.printStackTrace();
            }
        });
    }


    /**
     * 批量消费消息
     * 累计消息数量|累计消息大小|指定时间拉取一次满足其一即执行
     * @throws PulsarClientException
     */
    @GetMapping("/model/comsumerByBatch")
    public void comsumerByBatch() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .batchReceivePolicy(BatchReceivePolicy.builder()
                        .maxNumMessages(5) //累计消息数量，默认-1，不限制
                        .maxNumBytes(1024 * 1024)//累计消息大小，默认 10 * 1024 * 1024
                        .timeout(200, TimeUnit.MILLISECONDS)//指定时间拉取一次，默认 100
                        .build())
                .subscribe();
        Messages<byte[]> messages = consumer.batchReceive();
//        consumer.batchReceiveAsync();//异步同理
        System.out.println("本次接收"+messages.size()+"条");
        for (Message<byte[]> message : messages) {
            System.out.println(new String(message.getData()));
        }
        consumer.acknowledge(messages);//确认消息被消费
        consumer.close();
    }

    /**
     * 取消消费
     * @throws PulsarClientException
     */
    @GetMapping("negativeAcknowledgement")
    public void negativeAcknowledgement() throws PulsarClientException {
        PulsarClient pulsarFactory = pulsarConf.pulsarFactory();
        Consumer<byte[]> consumer = pulsarFactory.newConsumer()
                .topic("my-topic")
                .subscriptionName("my-subscription")
                .subscribe();
        Message<byte[]> receive = consumer.receive();
        MessageId messageId = receive.getMessageId();
        try {
            System.out.println(new String(receive.getData()));
            consumer.acknowledge(receive);//确认消息被消费
        } catch (PulsarClientException e) {
            consumer.negativeAcknowledge(messageId);//消息取消消费，会被重新消费
            e.printStackTrace();
        }
        consumer.close();
    }




}
